package com.gitee.huanminabc.jnetty.server;

import com.gitee.huanminabc.jcommon.base.SnowIdUtil;
import com.gitee.huanminabc.jcommon.multithreading.executor.SleepTools;
import com.gitee.huanminabc.jnetty.base.*;
import com.gitee.huanminabc.jnetty.server.channel.MyServerNewIdNioSocketChannel;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ResourceLeakDetector;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.function.Consumer;

@Slf4j
@Getter
@Setter
public class NettyServer {
    private Integer port;
    private String key = SnowIdUtil.uniqueLongHex(); //服务端唯一标识
    // 创建ServerBootstrap实例，服务器启动对象
    private final ServerBootstrap bootstrap = new ServerBootstrap();

    // 创建boss线程组，用于接收连接
    private final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    // 创建worker线程组，用于处理连接上的I/O操作，含有子线程NioEventGroup个数为CPU核数大小的2倍
    private final EventLoopGroup workerGroup = new NioEventLoopGroup();

    private Consumer<ChannelPipeline> channelPipelineConsumer;

    //channel对象数组,这个是会话级别

    private LogLevel logLevel = LogLevel.DEBUG; //日志级别
    private Integer so_backlog = 1024; //TCP缓冲区
    //是否开启心跳处理 ,需要客户端和服务端都开启,否则会出现客户端发送心跳,服务端不回复
    private boolean isCloseHeartbeatHandler = false;
    private int allHeartbeatTime = 360;//心跳时间,单位秒, 如果180秒内没有向客户端发送任何数据或者没有接收到客户端的任何数据,则触发一个事件.
    private int writeHeartbeatTime = 180;//心跳时间,单位秒, 如果180秒内没有向客户端发送任何数据,则触发一个事件.
    private int readHeartbeatTime = 180;//心跳时间,单位秒, 如果180秒内没有接收到客户端的任何数据,则触发一个事件.

    private HeartbeatHandlerEvent heartbeatHandlerEvent;

    //会话通道,ChannelGroup自动会移除关闭的channel
    private ChannelGroup clientChannelGroup;

    //开启内存泄露检测
    private boolean isOpenMemoryLeakDetection = false;

    //异步启动的回调
    private Consumer<ChannelFuture> appRunFutureConsumer;

    private ChannelFuture serverChannelFuture;

    public NettyServer(Integer port, Consumer<ChannelPipeline> consumer) {
        this.port = port;
        this.channelPipelineConsumer = consumer;
        this.clientChannelGroup = new MyDefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    }

    public void startSync() {
        start(false);
    }

    //必须配合waitServe()使用
    public void startAsync() {
        start(true);
    }

    //    服务端的启动是阻塞的
    private void start( boolean async) {
        try {
            if (isOpenMemoryLeakDetection) {
                //开启内存泄露检测
                ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
            }
            // 使用链式编程配置参数
            // 将boss线程组和worker线程组暂存到ServerBootstrap
            bootstrap.group(bossGroup, workerGroup);

            bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); //设置内存池
            // 设置启动参数，初始化服务器连接队列大小。服务端处理客户端连接请求是顺序处理，一个时间内只能处理一个客户端请求
            // 当有多个客户端同时来请求时，未处理的请求先放入队列中
            bootstrap.option(ChannelOption.SO_BACKLOG, so_backlog); //设置TCP缓冲区

            //检测连接是否有效
            bootstrap.childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE);
            //超时时间
            bootstrap.childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 60000);

            // 设置服务端Channel类型为NioServerSocketChannel作为通道实现
            bootstrap.channel(MyServerNewIdNioSocketChannel.class);
            bootstrap.handler(new LoggingHandler(logLevel)); //设置日志级别
            // 添加ServerHandler到ChannelPipeline，对workerGroup的SocketChannel（客户端）设置处理器
            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                //每次有新的连接进来时，都会调用这个方法
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    clientChannelGroup.add(socketChannel);//添加到channelGroup
                    ChannelPipeline p = socketChannel.pipeline();
                    if (isCloseHeartbeatHandler) {
                        p.addLast(new IdleStateHandler(readHeartbeatTime, writeHeartbeatTime, allHeartbeatTime));
                        p.addLast(new CustomHeartbeatHandler(heartbeatHandlerEvent));//心跳处理
                    }
                    channelPipelineConsumer.accept(p);
                }
            });
            // 绑定端口
            serverChannelFuture = bind(port);
            if (!async){
                serverChannelFuture.channel().closeFuture().sync();
                log.info("{}:服务{}关闭:", key, port);
            }
        } catch (Exception e) {
            log.error("{}:服务{}启动失败:{}", key, port, e.getMessage());
            throw new NettyBizException(e);
        } finally {
            if (!async){
                // 优雅地关闭boss线程组
                bossGroup.shutdownGracefully();
                // 优雅地关闭worker线程组
                workerGroup.shutdownGracefully();
            }
        }
    }

    private ChannelFuture bind(int port) throws InterruptedException {
        log.info("{}:服务端准备启动,监听端口:{}", key, port);
        ChannelFuture channelFuture = bootstrap.bind(port).sync();
        channelFuture.addListener((ChannelFuture future) -> {
            if (appRunFutureConsumer != null) {
                appRunFutureConsumer.accept(future);
            }
            if (future.isSuccess()) {
                log.info("{}:服务端启动成功，监听端口:{}", key, port);
            } else {
                log.error("{}:服务端{}启动失败:{}", key, port, future.cause().getMessage());
            }
        });
        return channelFuture;
    }

    //等待主线程结束 , 异步的方式自行控制什么时候等待, 但是需要在主线程结束前调用,且必须在start之后
    public void waitServe() {
        // 阻止主线程退出，直到服务关闭（可选）
        try {
            while (serverChannelFuture == null) {
                SleepTools.ms(10);
            }
            serverChannelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("异常中断:{}", e.getMessage());
        } finally {
            // 关闭事件循环组
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    //关闭服务
    public void close() {
        log.info("{}:服务端{}手动强制关闭自己", key, port);
        serverChannelFuture.channel().close();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

    //获取指定id的客户端
    public Channel findChannel(String clientId) {
        return clientChannelGroup.find(MyNioSocketChannel.newChannelId(clientId));
    }


    //给客户端发消息,所有客户端
    public void sendClientMessage(Object msg) {
        if (!clientChannelGroup.isEmpty()) {
            ChannelGroupFuture channelFutures = clientChannelGroup.writeAndFlush(msg);
            channelFutures.addListener(future -> {
                if (!future.isSuccess()) {
                    log.error("sendServerMessage-{}-服务端发送消息给客户端失败了,error:{},message:{}", key, future.cause().getMessage(), msg);
                }
            });
        } else {
            log.error("sendServerMessage-{}-服务端发送消息给客户端失败了,error:{},message:{}", key, "客户端连接不存在", msg);
        }
    }

    //给指定客户端发送消息

    /**
     * @param clientId 唯一标识LongText
     * @param msg
     */
    public void sendClientMessage(String clientId, Object msg) {
        Channel channel = clientChannelGroup.find(MyNioSocketChannel.newChannelId(clientId));
        if (channel != null && channel.isActive()) {
            ChannelFuture channelFuture = channel.writeAndFlush(msg);
            channelFuture.addListener(future -> {
                if (!future.isSuccess()) {
                    log.error("sendServerMessage-{}-服务端发送消息给客户端失败了,error:{},message:{}", key, future.cause().getMessage(), msg);
                }
            });
        } else {
            log.error("sendServerMessage-{}-服务端发送消息给客户端失败了,error:{},message:{}", key, "客户端连接不存在", msg);
        }
    }

    //关闭服务
    public static void shutdown() {

    }
}
